package actor.persistence

import actor.persistence.Calculator._
import akka.actor.{ActorLogging, Props}
import akka.persistence.{PersistentActor, RecoveryCompleted, SaveSnapshotFailure, SaveSnapshotSuccess, SnapshotOffer}

/**
 * 事件来源模式具体的工作原理原则上就是把使状态产生变化的事件按发生时间顺序持久化，而不是把当前整个状态存储起来.
 * 在恢复状态时把日志中这些事件按原来的时间顺序重演一遍回到原来的状态。但为了减少事件的长久积存和加快恢复（recovery）效率，
 * PersistentActor也可以在一个时间间隔后把完整的当前状态作为快照（snapshop）存储起来。这样在状态恢复时就可以最新的快照为起点，只需对日志中快照时间之后的事件进行重演。
 *
 * Created by DJJ on 2019/12/28.
 */
object Calculator {

  sealed trait Command

  case class Operand(x: Int) extends Command

  case class Add(x: Int) extends Command

  case class Sub(x: Int) extends Command

  case class Mul(x: Int) extends Command

  case class Div(x: Int) extends Command

  case class ShowResult(x: Double) extends Command

  case object BackupResult extends Command //saveSnapshot

  sealed trait Event

  case class SetNum(x: Int) extends Event

  case class Added(x: Int) extends Event

  case class Subtracted(x: Int) extends Event

  case class Multiplied(x: Int) extends Event

  case class Divided(x: Int) extends Event


  case class State(result: Int) {
    def updateState(evt: Event): State = evt match {
      case SetNum(x) => copy(result = x)
      case Added(x) => copy(result = this.result + x)
      case Subtracted(x) => copy(result = this.result - x)
      case Multiplied(x) => copy(result = this.result * x)
      case Divided(x) => copy(result = this.result / x)
    }
  }

  case class LogMessage(msg: String) //broadcase message type
  def props = Props(new Calculator)

}

class Calculator extends PersistentActor with ActorLogging {

  var state = State(0)
  val snapShotInterval = 5

  override def persistenceId: String = "persistence-actor"

  override def receiveCommand: Receive = {
    case Operand(x) => persist(SetNum(x))(handleEvent)
    case Add(x) => persist(Added(x))(handleEvent)
    case Sub(x) => persist(Subtracted(x))(handleEvent)
    case Mul(x) => persist(Multiplied(x))(handleEvent)
    case Div(x) if (x != 0) => persist(Divided(x))(handleEvent)
    case ShowResult =>
      println("--------------:"+state)
      context.system.eventStream.publish(LogMessage(s"Current state: $state"))
    case BackupResult =>
      saveSnapshot(state)
      context.system.eventStream.publish(LogMessage(s"Manual saving snapshot: $state"))
    case SaveSnapshotSuccess(metadata) =>
      context.system.eventStream.publish(LogMessage(s"Successfully saved state: $state"))
    case SaveSnapshotFailure(metadata, reason) =>
      context.system.eventStream.publish(LogMessage(s"Saving state: $state failed!"))

  }


  override def receiveRecover: Receive = {
    /**
     * 恢复过程中会从最近一次的快照 saveSnapshot 没保存的事件开始从新执行一遍已恢复到停止之前的状态
     * 比如 A B C D E F 这么多个事件都 persist 持久化,如果最近一次快照 saveSnapshot 保存了A B C D
     * 下次重启恢复则是从 E F开始执行
     */
    case evt: Event => {
      println("-------------------")
      state = state.updateState(evt)
      context.system.eventStream.publish(LogMessage(s"Restoring event: $evt"))
    }

    /**
      * 重启开始执行持久化数据恢复
     */
    case SnapshotOffer(mdata, sts: State) => {
      state = sts.copy(sts.result)
      context.system.eventStream.publish(LogMessage(s"Restoring snapshot: $mdata"))
    }
    //重播完成
    case RecoveryCompleted => log.info(s"Recovery completed with starting state: $state")
  }


  def handleEvent(evt: Event): Unit = { //update state and publish progress
    state = state.updateState(evt)
    context.system.eventStream.publish(LogMessage(s"Producing event: $evt"))
    if (lastSequenceNr % snapShotInterval == 0 && lastSequenceNr != 0) {
      saveSnapshot(state)
      context.system.eventStream.publish(LogMessage(s"Saving snapshot: $state after $snapShotInterval events"))
    }
  }

  override def onPersistRejected(cause: Throwable, event: Any, seqNr: Long): Unit = {
    log.info(s"Persistence Rejected: ${cause.getMessage}")
  }

  override def onPersistFailure(cause: Throwable, event: Any, seqNr: Long): Unit = {
    log.info(s"Persistence Error: ${cause.getMessage}")
  }
}